package org.liang.code.jms.queue;

import jakarta.jms.*;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory;
import org.liang.code.utils.jms.JMSCommons;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;

/**
 * 队列生产者
 */
@Slf4j
public class TProducer implements Runnable{

    private final CountDownLatch latch;
    private final ActiveMQJMSConnectionFactory factory;
    // 是否使用事务
    private final Boolean transactionMessage;

    public TProducer(CountDownLatch latch, ActiveMQJMSConnectionFactory factory, Boolean transactionMessage) {
        this.latch = latch;
        this.factory = factory;
        this.transactionMessage = transactionMessage;
    }

    @SneakyThrows
    @Override
    public void run() {

        JMSContext context = null;
        JMSProducer producer;

        try {
            // 创建JMSContext
            if (transactionMessage) {
                context = factory.createContext(Session.SESSION_TRANSACTED);
            }else {
                context = factory.createContext(Session.AUTO_ACKNOWLEDGE);
            }
            // 创建消息
            TextMessage message = context.createTextMessage("message id : " + JMSCommons.counter.getAndAdd(1) + ", haha!");
            // 配置唯一属性
            message.setStringProperty(Message.HDR_DUPLICATE_DETECTION_ID.toString(), UUID.randomUUID().toString());
            // 创建生产者
            producer = context.createProducer();
            // 消息5秒后丢弃
            // producer.setTimeToLive(5000);
            // 消息十秒可消费
            // producer.setDeliveryDelay(10000);
            // 发送消息
            producer.send(context.createQueue(JMSCommons.CORE_QUEUE), message);
            // 记录日志
            log.info("MessageID:" + message.getJMSMessageID() + "Message sent: " + message.getText());

            // 使用事务
            if (transactionMessage) {
                // 提交事务
                context.commit();
            }
        }catch (JMSException e) {
            if (transactionMessage) {
                // 回滚事务
                context.rollback();
                log.error("事务已回滚！", e);
            }
        }finally {
            // 关闭连接
            if (context != null) {
                context.close();
            }
            // 更新计数器
            latch.countDown();
        }
    }
}
